package com.example
import org.apache.spark.sql.ForeachWriter
import redis.clients.jedis.{Jedis, JedisPool, JedisPoolConfig}
object RedisWriter {
  val config = new JedisPoolConfig
  //最大连接数
  config.setMaxTotal(20)
  //最大空闲连接数
  config.setMaxIdle(10)
  //设置连接池属性分别有： 配置 主机名 端口号 连接超时时间
  val pool = new JedisPool(config, "node1", 6379, 10000)
  //连接池
  def getConnection(): Jedis = {
    pool.getResource
  }
}
class RedisWriter extends ForeachWriter[BusInfo] {
  var jedisPool: JedisPool = null
  var jedis: Jedis = null
  override def open(partitionId: Long, epochId: Long): Boolean = {
    jedis = RedisWriter.getConnection()
    true
  }
  override def process(value: BusInfo): Unit = {
    val lglat: String = value.lglat
    val deployNum: String = value.deployNum
    jedis.set(deployNum, lglat);
  }
  override def close(errorOrNull: Throwable): Unit = {
    jedis.close()
  }
}
